/**
 * 开发者   : Marco
 * 创建时间 : 2016年7月26日, 下午5:31
 */


#include "tpool.h"

#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>

#ifndef MARCO_NO_TPOOL
MARCOPP_NAMESPACE_BEGIN
namespace mt {
/**
 * 任务结构体
 */
typedef struct tpool_work {
    void (*workfunc)(void *); // 任务函数
    void *arg; // 传入任务函数的参数
    struct tpool_work *next; // 下一节点
} tpool_work_t; // 任务链表结构体,每个任务一个节点

/**
 * 实现类
 */
class tpool_impl {
public:
    /**
     * 构造
     * @param threads 最大工作线程数
     * @param stack_size 栈大小(M)
     */
    explicit tpool_impl(int threads, int stack_size) : head(NULL), tail(NULL) {
	pthread_mutex_init(&lock, NULL); // 互斥锁初始化
	pthread_cond_init(&cond, NULL); // 条件变量初始化

	int i;
	pthread_t tid;
	pthread_attr_t attr;
	size_t size = 1024 * 1024 * stack_size; // 栈大小,决定最大线程创建人数目,Linux默认是8M-10M

	pthread_attr_init(&attr); // 初始化线程属性
	pthread_attr_setstacksize(&attr, size); // 设置栈大小

	for (i = 0; i < threads; ++i) {
	    if (pthread_create(&tid, &attr, threadEntry, this) != 0) {
		break;
	    }
	}

	pthread_attr_destroy(&attr); // 销毁线程属性
    }
    
    /**
     * 添加任务
     * @param func 执行回调函数
     * @param arg 携带参数
     */
    void add_work(void (*func)(void *arg), void *arg) {
	tpool_work_t *work;

	work = (tpool_work_t *)malloc(sizeof (tpool_work_t));

	if (!work)
	    return; // 申请空间失败

	work->workfunc = func;
	work->arg = arg;
	work->next = NULL;

	pthread_mutex_lock(&lock); // 对链表操作,先加锁

	if (head == NULL) { // 头节点为空
	    head = work; // work为新的头节点

	} else {
	    tail->next = work; // 插到链表尾
	}

	tail = work; // 更新尾节点
	pthread_mutex_unlock(&lock); // 解锁
	pthread_cond_signal(&cond); // 唤醒一个线程
    }

private:
    /**
     * 工作线程入口, 静态
     * @param arg 线程池对象
     * @return 退出状态,未用到
     */
    static void* threadEntry(void *arg) {
	tpool_work_t *work; // 任务链表结构体指针
	tpool_impl* pool = (tpool_impl*)arg;

	// 线程进入循环处理任务或睡眠
	while (1) {
	    pthread_mutex_lock(&pool->lock); // 获得互斥锁

	    // 睡眠被唤醒时会得到锁, 检查任务链表头节点如果为空则继续睡眠.
	    while (!pool->head) {
		//printf("Thread %lu waiting......\n", pthread_self());
		// 进入睡眠状态,等待唤醒, 睡眠时先释放锁, 让其它线程可以得到
		pthread_cond_wait(&pool->cond, &pool->lock);
	    }

	    // 执行到此处, 表示此线程被唤醒,并得到互斥锁,且任务链表中有任务
	    work = pool->head; // 指向链表头
	    pool->head = pool->head->next; // 原头节点指向下一节点
	    pthread_mutex_unlock(&pool->lock); // 解锁,让其它线程可以得到锁, 即使被唤醒,而该任务已从链表中移走,也不会重复执行

	    //printf("Thread %lu working......\n", pthread_self());
	    work->workfunc(work->arg); // 执行任务函数
	    free(work); // 释放该节点空间
	}

	pthread_exit(NULL); // 执行不到
    }	

    // 私有成员变量
    tpool_work_t* head; // 任务链表头
    tpool_work_t* tail; // 任务链表尾

    // 实现取消
    pthread_mutex_t lock; // 互斥锁
    pthread_cond_t cond; // 条件变量
};


tpool::tpool(int threads, int stack_size) : impl(new tpool_impl(threads, stack_size)) { }
tpool::~tpool(void) { delete impl; }
void tpool::add_work(void (*func)(void *arg), void *arg) { impl->add_work(func, arg); }

} // namespace mt
MARCOPP_NAMESPACE_END
#endif // MARCO_NO_TPOOL